package com.gblfy.lykafka.consumer;

import com.alibaba.fastjson.JSON;
import com.gblfy.common.constant.KafkaTopicConstants;
import com.gblfy.common.entity.Order;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * Kafka消费者
 *
 * @author gblfy
 * @date 2021-09-28
 */
@Component
public class KafkaConsumer {
    private final static Logger log = LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(topics = {KafkaTopicConstants.KAFKA_MSG_TOPIC}, groupId = KafkaTopicConstants.KAFKA_MSG_TOPIC_GROUP)
    public void onMessage(ConsumerRecord<Integer, String> record) {
        String acceptJsonValue = record.value();
        log.info("消费者接收到消息主题:{} ,消息的分区:{} ,消息偏移量:{}  ,消息key: {}  ,消息values:{} ",
                record.topic(), record.partition(), record.offset(), record.key(), acceptJsonValue);

        //解析熊kafka中发送的消息
        Order order = JSON.parseObject(acceptJsonValue, Order.class);
        log.info("订单编码:{} ,订单号:{} ,订单创建时间:{}", order.getOrderId(), order.getOrderNum(), order.getCreateTime());
    }
}


